package com.atguigu.gmall.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Set;

/**
 * @Author lzc
 * @Date 2022/12/26 14:05
 */
public class DimApp extends BaseApp {
    public static void main(String[] args) {
        new DimApp().init(
            2001,
            2,
            "DimApp",
            Constant.TOPIC_ODS_DB
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
        // 执行对流操作的业务逻辑
        // 1. 对数据做 etl, 过滤掉脏数据
        SingleOutputStreamOperator<JSONObject> etledStream = etl(stream);
        
        // 2. 使用 flink sql cdc 读取 配置表的数据
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);
        
        // 3. 根据配置信息, 在 phoenix 中, 创建对应的维度表
        tpStream = createDimTable(tpStream);
        // 4. 每一条维度找到它对应的配置信息: 数据流 connect 配置流
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimAndTpStream = connect(etledStream, tpStream);
        // 5. 删除不需要的列
        dimAndTpStream = delNotNeedColumns(dimAndTpStream);
        
        // 6. 把不同维度数据写入到不同的dim表中
        writeToPhoenix(dimAndTpStream);
        
    }
    
    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimAndTpStream) {
        /*
        sink 到 phoenix:
        1.找到专门的 phoenix sink
            目前没有
        
        2. phoenix 写入可以使用 jdbc, 能否使用 jdbc sink
            jdbc sink 要求这个流中的数据只能写入到同一个表中
            dimAndTpStream 需要写入到多个不同的维度表中, 所以不能用 jdbc sink
            
        3. 自定义 sink
        
         */
        
        dimAndTpStream.addSink(FlinkSinkUtil.getPhoenixSink());
    }
    
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> delNotNeedColumns(
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dimAndTpStream) {
        return dimAndTpStream
            .map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
                @Override
                public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                    JSONObject data = t.f0;
                    List<String> columns = Arrays.asList(t.f1.getSinkColumns().split(","));
                    
                    Set<String> keys = data.keySet();
                    keys.removeIf(key -> !columns.contains(key) && !"op_type".equals(key));
                    return t;
                }
            });
    }
    
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(
        SingleOutputStreamOperator<JSONObject> dataStream,
        SingleOutputStreamOperator<TableProcess> tpStream) {
        // 1. 把配置流做成广播流
        // 广播状态如何保存?
        // key: string   表名:ALL
        // value: TableProcess
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpStateDesc", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);
        // 2. 数据流去 connect 广播流
        return dataStream
            .connect(tpBcStream)
            .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                // 4. 处理数据流中的数据的时候, 从广播状态获取配置信息
                @Override
                public void processElement(JSONObject obj,
                                           ReadOnlyContext ctx,
                                           Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                    ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                    
                    String key = obj.getString("table") + ":ALL";
                    TableProcess tp = state.get(key);
                    
                    // obj可能是维度数据,也可能是事实表数据, 所以 tp 有可能是空
                    if (tp != null) {
                        JSONObject data = obj.getJSONObject("data");
                        // 从 data 中无法得到这条数据是 insert 还是 update, 到后期需要对维度数据不同的操作, 进行不同的处理. 需要把 type 写入到 data 中
                        // 后期要用
                        data.put("op_type", obj.getString("type"));
                        out.collect(Tuple2.of(data, tp));
                        
                    }
                    
                }
                
                // 3. 把配置信息放入到广播状态
                @Override
                public void processBroadcastElement(TableProcess tp,
                                                    Context ctx,
                                                    Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                    // 获取广播状态
                    BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                    // 放入
                    String key = tp.getSourceTable() + ":" + tp.getSourceType();
                    
                    // 如果 op=d, 表示删除对应的配置信息, 这个时候,也应该删除广播状态中的数据
                    if ("d".equals(tp.getOp())) {
                        state.remove(key);
                    } else {
                        state.put(key, tp);
                    }
                }
            });
        
        
    }
    
    private SingleOutputStreamOperator<TableProcess> createDimTable(SingleOutputStreamOperator<TableProcess> tpStream) {
        // 只对维度表做操作
        return tpStream
            .filter(tp -> "dim".equals(tp.getSinkType()))  //
            .process(new ProcessFunction<TableProcess, TableProcess>() {
                
                private Connection conn;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    // 1. 建立到 phoenix 的链接
                    conn = JdbcUtil.getPhoenixConnection();
                }
                
                @Override
                public void close() throws Exception {
                    // 6. 关闭到 phoenix 的链接
                    JdbcUtil.closeConnection(conn);
                }
                
                @Override
                public void processElement(TableProcess tp,
                                           Context ctx,
                                           Collector<TableProcess> out) throws Exception {
                    String op = tp.getOp();
                    
                    if ("c".equals(op) || "r".equals(op)) {
                        // 建表
                        createTable(tp);
                    } else if ("d".equals(op)) {
                        // 配置信息删除了, 对应的维度表应该删除
                        delTable(tp);
                    } else {
                        // 配置信息发生了变化 u
                        // 1. 先删表
                        delTable(tp);
                        // 2. 再建表
                        createTable(tp);
                    }
                    
                    out.collect(tp);
                }
                
                // 删表
                private void delTable(TableProcess tp) throws SQLException {
                    // 删表操作
                    // 2. 拼接一个删表的 sql 语句
                    // drop table a
                    String sql = "drop table " + tp.getSinkTable();
                    // 3. 获取一个预处理语句
                    PreparedStatement ps = conn.prepareStatement(sql.toString());
                    ps.execute();
                    ps.close();
                }
                
                // 建表
                private void createTable(TableProcess tp) throws SQLException {
                    // 建表操作
                    // 2. 拼接一个建表的 sql 语句
                    // create table if not exists user(id varchar, name varchar, constraint pk primary key(id)) SALT_BUCKETS = 4
                    StringBuffer sql = new StringBuffer();
                    sql
                        .append("create table if not exists ")
                        .append(tp.getSinkTable())
                        .append("(")
                        // 拼接字段
                        .append(tp.getSinkColumns().replaceAll("[^,]+", "$0 varchar"))
                        .append(",constraint pk primary key(")
                        .append(tp.getSinkPk() == null ? "id" : tp.getSinkPk())
                        .append("))")
                        .append(tp.getSinkExtend() == null ? "" : tp.getSinkExtend());
                    System.out.println("phoenix 建表语句: " + sql);
                    // 3. 获取一个预处理语句
                    PreparedStatement ps = conn.prepareStatement(sql.toString());
                    // 4. 执行建表语句
                    ps.execute();
                    // 5. 关闭预处理语句
                    ps.close();
                }
            });
    }
    
    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        // cdc: 程序已启动, 默认t先读取所有的数据(快照),然后根据 binlog 读取变化的数据
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("hadoop162")
            .port(3306)
            .databaseList("gmall_config") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
            .tableList("gmall_config.table_process") // set captured table
            .username("root")
            .password("aaaaaa")
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();
        
        return env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc")
            .map(new MapFunction<String, TableProcess>() {
                @Override
                public TableProcess map(String value) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    String op = obj.getString("op");
                    
                    TableProcess tp;
                    if ("d".equals(op)) {
                        // 如果是删除操作, 则获取 before
                        tp = obj.getObject("before", TableProcess.class);
                    } else {
                        // u c r 获取 after
                        tp = obj.getObject("after", TableProcess.class);
                        
                    }
                    tp.setOp(op); // 把 op 的值, 设置到对象中
                    return tp;
                }
            });
        
        
    }
    
    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream
            .filter(new FilterFunction<String>() {
                @Override
                public boolean filter(String value) throws Exception {
                    try {
                        JSONObject obj = JSON.parseObject(value);
                        String type = obj.getString("type");
                        String data = obj.getString("data");
                        // 1. 库必须是 gmall2022 2. table not null 3. type: insert 或者 update 4.data如果当做一个 string 来处理, 长度 > 2
                        return "gmall2022".equals(obj.getString("database"))
                            && null != obj.getString("table")
                            && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type))
                            && null != data
                            && data.length() > 2;
                    } catch (Exception e) {
                        System.out.println("数据不是合法的 json 数据: " + value);
                        // 解析 json 格式数据出错,表示不是 json 格式
                        return false;
                    }
                    
                }
            })
            .map(json -> JSON.parseObject(json.replace("bootstrap-", "")));
    }
}
/*
业务数据已经到了 ods 层: ods_db
    事实表
    维度表

Dim 层 App 的任务:
    消费 ods_db, 过滤出其中需要的维度表的数据, 然后写入到 Phoenix 中
    
    在 mysql 中的一张维度表就对应着 phoenix 中的一张维度表
    
    
maxwell:
    处理历史数据 maxwell-bootstrap, 同步历史数据

cdc:

op: 读取快照的时候   r
        before=null after有值
    update        u
        before 有值  after 有值
    delete        d
        before 有值 after=null
    insert        c
        before=null  after 有值
    更新的主键:
        先 d 再 c
        
   
   当配置信息中有一条数据, 意味着,要在 phoenix 中创建一张表
    r c 建表
    d   删表
    u   先删,再建
   

phoenix:
    database
    
hbase
    namespace
    
---------------------
 SALT_BUCKETS = 4
    用来在 phoenix 中建盐表
    
hbase:
 HRegionServer 服务器进程, 存 Region
 
 Region 存储数据
 
 当在 hbase 中,建表,默认一张表有一个 Region.
 最初, 所有数据都在同一个 Region, 在同一个HRegionServer
 
 随着 Region 的增长, Region 会自动分裂, 一分为二
 旧: 10G
 新: min(? M, 10g)
 
 当同一张表的 Region 分裂成多个之后, HBase 会把 Region 迁移不同的节点上
 
 
生成环境下, 为了避免分裂和迁移 一般会建预分区表£

phoenix 建立预分区表?
    盐表
    
-------
yanr 提交:
  session
     bin/yarn-session.sh -d
     bin/flink run -d \
     -c com.atguigu.gmall.realtime.app.dim.DimApp \
     /opt/gmall220730/gmall-realtime-1.0-SNAPSHOT.jar
  per-job
    bin/flink run -d \
    -t yarn-per-job \
    -Dclassloader.check-leaked-classloader=false \
    -c com.atguigu.gmall.realtime.app.dim.DimApp \
    /opt/gmall220730/gmall-realtime-1.0-SNAPSHOT.jar
  
  application
    bin/flink run-application \
    -t yarn-application \
    -Dclassloader.resolve-order=parent-first \
    -c com.atguigu.gmall.realtime.app.dim.DimApp \
    /opt/gmall220730/gmall-realtime-1.0-SNAPSHOT.jar
  
flink 通过 cdc连接 mysql 的 连不上, 极有可能是 jdk 的版本问题, 1.8.3xxx.
解决:把你 windows 的 jdk 版本换成老师给的那个版本

DimApp


yarnd 三种部署模式:
1. yarn-session(生成环境用的比较少)
	需要先启一个集群, 然后, 所有的 job 运行在这一个集群上
2. yarn-per-job
	每运行一个 job, 就启动一个集群  (从 1.15 开始, 官方标记为过时. 用这个比较多)
		main 函数在客户端执行
3. yarn-application (1.12 新增)  (官方建议用这个)
	每运行一个 job, 就启动一个集群
		main 是在集群上执行(JobManager(AM))
    
 
 */
